package com.xiaoyushu.count.biz.consumer;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.xiaoyushu.count.biz.constant.MQConstants;
import com.xiaoyushu.count.biz.domain.mapper.NoteCountDOMapper;
import com.xiaoyushu.count.biz.domain.mapper.UserCountDOMapper;
import com.xiaoyushu.count.biz.model.dto.AggregationCountLikeUnlikeNoteMqDTO;
import com.xiaoyushu.framework.common.util.JsonUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionTemplate;

import java.util.List;

@Component
@RocketMQMessageListener(consumerGroup = "xiaoyushu_group_"+ MQConstants.TOPIC_COUNT_NOTE_LIKE_2_DB,//Group 组
        topic = MQConstants.TOPIC_COUNT_NOTE_LIKE_2_DB//主题Topic
)
@Slf4j
public class CountNoteLike2DBCousumer implements RocketMQListener<String> {


    @Resource
    private NoteCountDOMapper noteCountDOMapper;

    //每秒创建 5000 个令牌
   private RateLimiter rateLimiter= RateLimiter.create(5000);
    @Autowired
    private TransactionTemplate transactionTemplate;
    @Autowired
    private UserCountDOMapper userCountDOMapper;

    /**
     * @param body
     */
    @Override
    public void onMessage(String body) {
        //流量削峰L通过获取令牌,如果没有令牌可用,将阻塞,直到获得
        rateLimiter.acquire();
        log.info("## 消费到了 MQ [计数:笔记点赞数入库],{}..",body);

       // Map<Long,Integer> countMap=null;
        List<AggregationCountLikeUnlikeNoteMqDTO> countList=null;
        try {
            countList= JsonUtils.parseList(body, AggregationCountLikeUnlikeNoteMqDTO.class);
        }catch (Exception e){
            log.error("## 解析 JSON 字符串异常",e);
        }

        if (CollUtil.isNotEmpty(countList)){
            // 判断数据库中 t_note_count 表，若笔记计数记录不存在，则插入；若记录已存在，则直接更新
          countList.forEach(item->{
              Long creatorId = item.getCreatorId();
              Long noteId = item.getNoteId();
              Integer count = item.getCount();
              //编程式事务,保证两条语句的原子性
              transactionTemplate.execute(status -> {
                  try {
                      noteCountDOMapper.insertOrUpdateLikeTotalByNoteId(count, noteId);
                      userCountDOMapper.insertOrUpdateLikeTotalByUserId(count, creatorId);
                      return true;
                  }catch (Exception e){
                      status.setRollbackOnly();//标记事务为回滚
                      log.error("",e);
                  }
                  return false;
              });
          });

        }

    }
}
